package cn.demoncat.util.redis.util;

import cn.demoncat.util.lang.LocalDateUtil;
import cn.demoncat.util.lang.ThreadUtil;
import cn.demoncat.util.redis.config.RedisIsolator;
import cn.demoncat.util.redis.entity.RedisDelay;
import cn.demoncat.util.redis.entity.RedisLock;
import cn.demoncat.util.mybatis.util.TransactionUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;

/**
 * Redis 延时队列工具
 *
 * 注：为了跟踪项目中使用了哪些延时队列，必须使用RedisDelay对象来声明延时队列的缓存常量
 *
 * @author 延晓磊
 * @since 2021年02月23日
 */
@Component
public class RedisDelayUtil {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonUtil redissonUtil;
    @Resource
    private RedisIsolator redisIsolator;

    // 消费锁：多线程交替消费队列
    private static final RedisLock LOCK = new RedisLock("DELAY:");

    private static final Logger LOG = LoggerFactory.getLogger(RedisDelayUtil.class);

    /**
     * 添加/修改
     *
     * @param key
     * @param value
     * @param date  到期时间
     * @return 是否覆盖
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public boolean add(String key, String value, LocalDateTime date){
       return set(key).add(value, LocalDateUtil.getTime(date));
    }

    /**
     * 添加/修改
     *
     * @param key
     * @param value
     * @param seconds   到期秒数
     * @return 是否覆盖
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public boolean add(String key, String value, int seconds){
       return set(key).add(value, System.currentTimeMillis() + seconds * 1000);
    }

    /**
     * 添加/修改（事务执行后）
     *
     * @param key
     * @param value
     * @param date  到期时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public void txAdd(String key, String value, LocalDateTime date){
        TransactionUtil.afterCommit(() -> add(key, value, date));
    }

    /**
     * 添加/修改（事务执行后）
     *
     * @param key
     * @param value
     * @param seconds   到期秒数
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public void txAdd(String key, String value, int seconds){
        TransactionUtil.afterCommit(() -> add(key, value, seconds));
    }

    /**
     * 获取
     *
     * @param key
     * @param min   起始时间戳
     * @param max   结束时间戳
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Set<String> get(String key, long min, long max){
        return set(key).rangeByScore(min, max);
    }

    /**
     * 获取
     *
     * @param key
     * @param max   结束时间戳
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Set<String> get(String key, long max){
        return get(key, 0, max);
    }

    /**
     * 获取
     *
     * @param key
     * @param min   起始时间
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Set<String> get(String key, LocalDateTime min, LocalDateTime max){
        return set(key).rangeByScore(min == null ? 0 : LocalDateUtil.getTime(min), max == null ? LocalDateUtil.getTime(LocalDateTime.MAX) : LocalDateUtil.getTime(max));
    }

    /**
     * 获取
     *
     * @param key
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Set<String> get(String key, LocalDateTime max){
        return get(key, null, max);
    }

    /**
     * 获取全部
     * 
     * @param key
     * @return
     * 
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Set<String> get(String key){
        return set(key).range(0, -1);
    }

    /**
     * 获取（包含到期时间）
     *
     * @param key
     * @param min   起始时间戳
     * @param max   结束时间戳
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Map<String, LocalDateTime> getTime(String key, long min, long max){
        Map<String, LocalDateTime> map = new HashMap<>();
        for (ZSetOperations.TypedTuple<String> entry : Objects.requireNonNull(set(key).rangeByScoreWithScores(min, max))) {
            map.put(entry.getValue(), LocalDateUtil.toLocalDateTime(Objects.requireNonNull(entry.getScore()).longValue()));
        }
        return map;
    }

    /**
     * 获取（包含到期时间）
     *
     * @param key
     * @param max   结束时间戳
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Map<String, LocalDateTime> getTime(String key, long max){
        return getTime(key, 0, max);
    }

    /**
     * 获取（包含到期时间）
     *
     * @param key
     * @param min   起始时间
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Map<String, LocalDateTime> getTime(String key, LocalDateTime min, LocalDateTime max){
        Map<String, LocalDateTime> map = new HashMap<>();
        for (ZSetOperations.TypedTuple<String> entry : Objects.requireNonNull(set(key).rangeByScoreWithScores(min == null ? 0 : LocalDateUtil.getTime(min), max == null ? LocalDateUtil.getTime(LocalDateTime.MAX) : LocalDateUtil.getTime(max)))) {
            map.put(entry.getValue(), LocalDateUtil.toLocalDateTime(Objects.requireNonNull(entry.getScore()).longValue()));
        }
        return map;
    }

    /**
     * 获取（包含到期时间）
     *
     * @param key
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Map<String, LocalDateTime> getTime(String key, LocalDateTime max){
        return getTime(key, null, max);
    }

    /**
     * 获取全部（包含到期时间）
     * @param key
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Map<String, LocalDateTime> getTime(String key){
      Map<String, LocalDateTime> map = new HashMap<>();
      for (ZSetOperations.TypedTuple<String> entry : Objects.requireNonNull(set(key).rangeWithScores(0, -1))) {
          map.put(entry.getValue(), LocalDateUtil.toLocalDateTime(Objects.requireNonNull(entry.getScore()).longValue()));
      }
      return map;
    }

    /**
     * 删除
     *
     * @param key
     * @param value
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, String value){
        return set(key).remove(value);
    }

    /**
     * 删除
     *
     * @param key
     * @param values
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, Set<String> values){
       return set(key).remove(values.toArray(new Object[0]));
    }

    /**
     * 删除
     *
     * @param key
     * @param min
     * @param max
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, long min, long max){
        return set(key).removeRangeByScore(min, max);
    }

    /**
     * 删除
     *
     * @param key
     * @param max
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, long max){
        return del(key, 0, max);
    }

    /**
     * 删除
     *
     * @param key
     * @param min   起始时间
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, LocalDateTime min, LocalDateTime max){
        return set(key).removeRangeByScore(min == null ? 0 : LocalDateUtil.getTime(min), max == null ? LocalDateUtil.getTime(LocalDateTime.MAX) : LocalDateUtil.getTime(max));
    }

    /**
     * 删除
     *
     * @param key
     * @param max   结束时间
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    public Long del(String key, LocalDateTime max){
        return del(key, null, max);
    }

    /**
     * 消费
     *
     * @param delay
     * @param consumer  消费者
     */
    public void consume(RedisDelay delay, Consumer<String> consumer){
        consume(delay.getKey(), delay.getExp(), consumer);
    }

    /**
     * 消费
     *
     * @param key
     * @param seconds   休眠秒数（间隔拉取）
     * @param consumer  消费者
     */
    public void consume(String key, int seconds, Consumer<String> consumer){
        ThreadUtil.loop("RedisDelayQueueConsumeThread - " + key, seconds, 0, () ->{
            // 加锁
            redissonUtil.tryLock(LOCK.build(key), () ->{
                // 查询小于当前的数据
                Set<String> values = get(key, System.currentTimeMillis());
                if (CollectionUtils.isNotEmpty(values)) {
                    // 防止异常时未删除已消费的数据，导致下次执行时重复消费，所以逐个处理和删除
                    for (String value : values) {
                        try {
                            // 消费
                            consumer.accept(value);
                            // 删除
                            del(key, value);
                        }catch (Exception e){
                            // 异常继续
                            LOG.warn("Redis Delay Queue 消费异常：" + value, e);
                        }
                    }
                }
            });
        });
    }


    /**
     * 获取Redis ZSet
     *
     * <pre>
     *     元素个数 ZCARD key
     *     删除元素 ZREM key e
     *     查询元素 ZRANGE key 0 -1
     *     查看分值 ZSCORE key e
     *     添加元素 ZADD key score e
     * </pre>
     *
     * @param key
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月10日
     */
    private BoundZSetOperations<String, String> set(String key){
        return stringRedisTemplate.boundZSetOps(redisIsolator.getKey(key));
    }
}
